use rustcloud_core::{Request, ServiceInstance, ServiceRegistry, LoadBalancerStrategy, TransportConfig};
use rustcloud_registry::InMemoryRegistry;
use rustcloud_loadbalancer::LoadBalancerManager;
use rustcloud_transport::HttpTransport;
use rustcloud_proxy::ServiceProxy;
use rustcloud_resilience::{RetryExecutor, RetryConfig, BackoffStrategy, CircuitBreaker, CircuitBreakerConfig};
use rustcloud_observability::{HealthCheckManager, MetricsCollector, AlertManager};
use rustcloud_bus::memory::InMemoryMessageBus;
use rustcloud_stream::{BusStreamSource, BusStreamSink, MapProcessor, FilterProcessor};
use rustcloud_admin::{ServiceDiscovery, LoadBalancerConfig, ServiceGateway};

use std::sync::Arc;
use std::time::Duration;
use tokio::time::timeout;
use wiremock::{MockServer, Mock, ResponseTemplate};
use wiremock::matchers::{method, path};
use serde_json::json;

/// 完整的微服务框架集成测试
/// 测试所有核心模块之间的协同工作
#[tokio::test]
async fn test_complete_microservice_framework_integration() {
    // 1. 初始化所有核心组件
    
    // 注册中心
    let registry = Arc::new(InMemoryRegistry::new());
    
    // 负载均衡器
    let load_balancer = Arc::new(LoadBalancerManager::new(LoadBalancerStrategy::RoundRobin));
    
    // 传输层
    let transport = Arc::new(HttpTransport::new());
    
    // 容错组件
    let retry_config = RetryConfig {
        max_attempts: 3,
        timeout: Duration::from_secs(5),
        backoff: BackoffStrategy::Exponential {
            initial_delay: Duration::from_millis(100),
            max_delay: Duration::from_secs(2),
            multiplier: 2.0,
        },
        retry_conditions: vec![],
        budget_percent: 10.0,
        jitter_enabled: true,
    };
    let retry_executor = Arc::new(RetryExecutor::new(retry_config));
    
    let circuit_breaker_config = CircuitBreakerConfig::default();
    let circuit_breaker = Arc::new(CircuitBreaker::new(circuit_breaker_config));
    
    // 可观测性组件
    let health_check_manager = Arc::new(HealthCheckManager::new());
    let metrics_collector = Arc::new(MetricsCollector::new());
    let alert_manager = Arc::new(AlertManager::new());
    
    // 消息总线
    let message_bus = Arc::new(InMemoryMessageBus::new().await.unwrap());
    
    // 服务发现
    let service_discovery = Arc::new(ServiceDiscovery::new(
        "test-discovery".to_string(),
        registry.clone(),
        Duration::from_secs(30),
    ));
    
    // 2. 设置模拟服务
    let server1 = MockServer::start().await;
    let server2 = MockServer::start().await;
    
    // 配置模拟服务响应
    Mock::given(method("POST"))
        .and(path("/api/user-service/get_user"))
        .respond_with(ResponseTemplate::new(200).set_body_json(json!({
            "user_id": 123,
            "name": "John Doe",
            "email": "john@example.com"
        })))
        .mount(&server1)
        .await;
    
    Mock::given(method("POST"))
        .and(path("/api/order-service/create_order"))
        .respond_with(ResponseTemplate::new(201).set_body_json(json!({
            "order_id": "ORD-001",
            "status": "created",
            "amount": 99.99
        })))
        .mount(&server2)
        .await;
    
    // 3. 注册服务实例
    let user_service_uri = server1.uri();
    let order_service_uri = server2.uri();
    
    let extract_host_port = |uri: &str| -> (String, u16) {
        let parts: Vec<&str> = uri.split(':').collect();
        let host = parts[1].trim_start_matches("//").to_string();
        let port = parts[2].parse().unwrap();
        (host, port)
    };
    
    let (user_host, user_port) = extract_host_port(&user_service_uri);
    let (order_host, order_port) = extract_host_port(&order_service_uri);
    
    let user_instance = ServiceInstance::new("user-service", &user_host, user_port, "http");
    let order_instance = ServiceInstance::new("order-service", &order_host, order_port, "http");
    
    registry.register(user_instance).await.unwrap();
    registry.register(order_instance).await.unwrap();
    
    // 4. 启动健康检查
    health_check_manager.start().await.unwrap();
    
    // 5. 配置流处理管道
    let mut stream_source = BusStreamSource::with_defaults(
        message_bus.clone(),
        "events.*",
    ).await.unwrap();
    
    let mut stream_sink = BusStreamSink::with_defaults(
        message_bus.clone(),
        "processed.events",
    ).await.unwrap();
    
    let event_processor = MapProcessor::with_mapper(|mut msg| {
        let mut payload = msg.payload().clone();
        payload["processed_at"] = json!(chrono::Utc::now().timestamp());
        payload["framework"] = json!("rustcloud");
        msg.set_payload(payload);
        msg
    });
    
    let filter_processor = FilterProcessor::with_predicate(|msg| {
        msg.payload().get("valid").and_then(|v| v.as_bool()).unwrap_or(true)
    });
    
    stream_source.start().await.unwrap();
    
    // 6. 创建服务代理进行端到端测试
    let proxy = ServiceProxy::new(
        registry.clone(),
        load_balancer.clone(),
        transport.clone(),
        circuit_breaker.clone(),
    );
    
    // 7. 执行业务流程测试
    
    // 7.1 测试用户服务调用
    let mut user_request = Request::new("user-service", "POST", "/api/user-service/get_user");
    user_request.set_body(json!({"user_id": 123}).to_string().into_bytes());
    
    let transport_config = TransportConfig::default();
    let user_response = timeout(
        Duration::from_secs(10),
        proxy.call(user_request, &transport_config)
    ).await.unwrap().unwrap();
    
    assert_eq!(user_response.status, 200);
    let user_body: serde_json::Value = serde_json::from_slice(
        &user_response.body.unwrap()
    ).unwrap();
    assert_eq!(user_body["user_id"], 123);
    assert_eq!(user_body["name"], "John Doe");
    
    // 7.2 测试订单服务调用
    let mut order_request = Request::new("order-service", "POST", "/api/order-service/create_order");
    order_request.set_body(json!({
        "user_id": 123,
        "items": [{"product": "Widget", "price": 99.99}]
    }).to_string().into_bytes());
    
    let order_response = timeout(
        Duration::from_secs(10),
        proxy.call(order_request, &transport_config)
    ).await.unwrap().unwrap();
    
    assert_eq!(order_response.status, 201);
    let order_body: serde_json::Value = serde_json::from_slice(
        &order_response.body.unwrap()
    ).unwrap();
    assert_eq!(order_body["order_id"], "ORD-001");
    assert_eq!(order_body["status"], "created");
    
    // 7.3 测试流处理管道
    let publisher = message_bus.create_publisher().unwrap();
    
    // 发布测试事件
    let test_event = rustcloud_bus::prelude::MessageBuilder::new()
        .topic("events.user_action")
        .payload(json!({
            "user_id": 123,
            "action": "order_created",
            "order_id": "ORD-001",
            "valid": true
        }))
        .build();
    
    publisher.publish(test_event).await.unwrap();
    
    // 处理流事件
    if let Some(Ok(message)) = timeout(
        Duration::from_secs(5),
        stream_source.next()
    ).await.unwrap() {
        // 应用过滤器
        if filter_processor.should_process(&message) {
            // 处理消息
            let processed = event_processor.process(message).await.unwrap();
            
            // 验证处理结果
            assert!(processed.payload().get("processed_at").is_some());
            assert_eq!(processed.payload()["framework"], "rustcloud");
            assert_eq!(processed.payload()["user_id"], 123);
            
            // 发送到输出
            stream_sink.send(processed).await.unwrap();
            stream_sink.flush().await.unwrap();
        }
    }
    
    // 8. 测试容错机制
    
    // 8.1 测试重试机制
    let retry_operation = || async {
        // 模拟可能失败的操作
        if fastrand::f64() < 0.7 {  // 70% 失败率
            Err(rustcloud_resilience::ResilienceError::operation_failed("Network timeout".to_string()))
        } else {
            Ok("Success".to_string())
        }
    };
    
    let retry_result = retry_executor.execute(retry_operation).await;
    // 重试最终应该成功或者返回错误
    assert!(retry_result.is_ok() || retry_result.is_err());
    
    // 9. 验证可观测性功能
    
    // 9.1 检查健康状态
    let health_status = health_check_manager.check_health().await;
    assert!(health_status.is_ok());
    
    // 9.2 收集指标
    metrics_collector.record_counter("test.requests", 1.0, vec![]).await;
    metrics_collector.record_histogram("test.latency", 50.0, vec![]).await;
    
    let metrics = metrics_collector.get_metrics().await;
    assert!(!metrics.is_empty());
    
    // 10. 清理资源
    stream_source.stop().await.unwrap();
    stream_sink.close().await.unwrap();
    health_check_manager.stop().await.unwrap();
    
    println!("✅ 完整的微服务框架集成测试通过！");
    println!("测试覆盖了以下模块的协同工作：");
    println!("  - rustcloud-core: 核心抽象和接口");
    println!("  - rustcloud-registry: 服务注册发现");
    println!("  - rustcloud-loadbalancer: 负载均衡");
    println!("  - rustcloud-transport: 网络传输");
    println!("  - rustcloud-proxy: 服务代理");
    println!("  - rustcloud-resilience: 容错机制");
    println!("  - rustcloud-observability: 可观测性");
    println!("  - rustcloud-bus: 消息总线");
    println!("  - rustcloud-stream: 流处理");
    println!("  - rustcloud-admin: 服务管理");
}

/// 测试高并发场景下的模块协同工作
#[tokio::test]
async fn test_high_concurrency_integration() {
    let registry = Arc::new(InMemoryRegistry::new());
    let load_balancer = Arc::new(LoadBalancerManager::new(LoadBalancerStrategy::RoundRobin));
    let transport = Arc::new(HttpTransport::new());
    let circuit_breaker_config = CircuitBreakerConfig::default();
    let circuit_breaker = Arc::new(CircuitBreaker::new(circuit_breaker_config));
    
    // 设置模拟服务
    let server = MockServer::start().await;
    Mock::given(method("POST"))
        .and(path("/api/test"))
        .respond_with(ResponseTemplate::new(200).set_body_string("OK"))
        .mount(&server)
        .await;
    
    let server_uri = server.uri();
    let (host, port) = {
        let parts: Vec<&str> = server_uri.split(':').collect();
        let host = parts[1].trim_start_matches("//").to_string();
        let port = parts[2].parse().unwrap();
        (host, port)
    };
    
    let instance = ServiceInstance::new("test-service", &host, port, "http");
    registry.register(instance).await.unwrap();
    
    let proxy = ServiceProxy::new(registry, load_balancer, transport, circuit_breaker);
    
    // 并发请求测试
    let mut handles = vec![];
    let proxy = Arc::new(proxy);
    
    for i in 0..50 {
        let proxy_clone = proxy.clone();
        let handle = tokio::spawn(async move {
            let mut request = Request::new("test-service", "POST", "/api/test");
            request.set_body(format!("request-{}", i).into_bytes());
            
            let transport_config = TransportConfig::default();
            proxy_clone.call(request, &transport_config).await
        });
        handles.push(handle);
    }
    
    // 等待所有请求完成
    let mut success_count = 0;
    for handle in handles {
        match handle.await.unwrap() {
            Ok(response) => {
                assert_eq!(response.status, 200);
                success_count += 1;
            }
            Err(e) => {
                eprintln!("Request failed: {}", e);
            }
        }
    }
    
    // 至少90%的请求应该成功
    assert!(success_count >= 45, "成功请求数: {}, 期望至少45", success_count);
    
    println!("✅ 高并发集成测试通过！成功处理 {}/50 个请求", success_count);
}

/// 测试错误场景下的系统恢复能力
#[tokio::test]
async fn test_error_recovery_integration() {
    let registry = Arc::new(InMemoryRegistry::new());
    let load_balancer = Arc::new(LoadBalancerManager::new(LoadBalancerStrategy::RoundRobin));
    let transport = Arc::new(HttpTransport::new());
    
    // 配置快速熔断的熔断器
    let mut circuit_breaker_config = CircuitBreakerConfig::default();
    circuit_breaker_config.failure_threshold = 3;
    circuit_breaker_config.timeout = Duration::from_millis(100);
    
    let circuit_breaker = Arc::new(CircuitBreaker::new(circuit_breaker_config));
    
    // 设置总是失败的模拟服务
    let server = MockServer::start().await;
    Mock::given(method("POST"))
        .and(path("/api/failing"))
        .respond_with(ResponseTemplate::new(500).set_body_string("Internal Server Error"))
        .mount(&server)
        .await;
    
    let server_uri = server.uri();
    let (host, port) = {
        let parts: Vec<&str> = server_uri.split(':').collect();
        let host = parts[1].trim_start_matches("//").to_string();
        let port = parts[2].parse().unwrap();
        (host, port)
    };
    
    let instance = ServiceInstance::new("failing-service", &host, port, "http");
    registry.register(instance).await.unwrap();
    
    let proxy = ServiceProxy::new(registry, load_balancer, transport, circuit_breaker);
    
    // 连续发送失败请求，触发熔断器
    for i in 0..5 {
        let mut request = Request::new("failing-service", "POST", "/api/failing");
        request.set_body(format!("request-{}", i).into_bytes());
        
        let transport_config = TransportConfig::default();
        let result = proxy.call(request, &transport_config).await;
        
        // 前几个请求应该是服务器错误，后面的请求应该是熔断器开路
        match result {
            Ok(response) => assert_eq!(response.status, 500),
            Err(_) => {
                // 熔断器已经开路，直接快速失败
                if i >= 3 {
                    println!("✅ 熔断器在第{}次请求后开路", i + 1);
                }
            }
        }
    }
    
    println!("✅ 错误恢复集成测试通过！熔断器正常工作");
}